package com.qingguo.Test

import java.io.PrintWriter
import java.security.MessageDigest

import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FileSystem, Path}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{DataFrame, SQLContext}

import scala.reflect.io.Path

import scala.io.Source
import java.io.File

object MD5Test {

  /**
    * (0,1@!@张三@!@23,6512bd43d9caa6e02c990b0a82652dca)
    * (1,2@!@李四@!@21,c20ad4d76fe97759aa27a0c99bff6710)
    *
    * 对每行进行加密：10李六20——>75be6a4edc94d79696da39f50a0f6cbc
    *
    */

  def main(args: Array[String]): Unit = {

    val tableName = "S02_DEMAND_MANAGE"

    val count = 5

    val whereExp = " "

    val conf = new SparkConf().setMaster("local").setAppName("MD5")
    val sc = new SparkContext(conf)

    conf.set("spark.sql.shuffle.partitions", "10")

    // spark  sql 上下文对象  spark sql的入口
    val spark = new SQLContext(sc)

    //读取json文件创建df  ,json文件每一行必须是一个对象
    val hiveDF: DataFrame = spark.read.json("spark/data/hive.json")

    //注册成表S02_DEMAND_MANAGE
    hiveDF.registerTempTable(tableName)

    //    [age: bigint, id: string, md5_coding: string]
    //    val sqlFullData = sqlContext.sql("SELECT * from " + tableName + whereExp)

    //res:[md5_coding: string]
    val sqlFullData: DataFrame = spark.sql("SELECT * from " + tableName + whereExp).select("md5_coding")
    //    val sqlFullData: DataFrame = spark.sql("SELECT * from " + tableName + whereExp)


    val md5 = new Md5Verifier

    /**
      * +-----------+-----------+-----------+--------+---+--------------------+----------------+---------+
      * |demand_code|demand_desc|demand_name|      dt| id|          md5_coding|person_in_charge|work_load|
      * +-----------+-----------+-----------+--------+---+--------------------+----------------+---------+
      * |     需求号|    需求描述|    需求名称|20201001|  1|fa68bce8e859485fx...|        需求负责|        4|
      * +-----------+-----------+-----------+--------+---+--------------------+----------------+---------+
      */

    //fa68bce8e859485f327504bbcd05010d
    val csmd5 = md5.md5Verify("需求号需求描述需求名称1需求负责4")

    println(csmd5)

    val inFileRDD = spark.sparkContext.textFile("spark/data/Md5data")

    val inFileMD5RDD = inFileRDD
      .zipWithIndex()
      .map(x => {
        //  mkString把集合转成字符串
        //  data:1张三23为每一行的数据
        val data = x._1.split("@!@").map(s => md5.replaceDouble(s)).mkString

        //        data:1需求号需求名称需求描述4需求负责

        val md5Val = md5.md5Verify(data)
        //      println("**************************************** md5Val=" + md5Val)
        (x._2, x._1, md5Val) // idx, data, md5_coding
      })

    inFileMD5RDD.foreach(println)

    var limitCnt = inFileMD5RDD.count().toInt

    //    //limitCnt=limit
    //    if (count.toLong < limitCnt) {
    //      limitCnt = count.toInt
    //    }
    //
    //导入隐式转换
    import spark.implicits._

    val (colIdx, colData, colMd5) = ("idx", "data", "md5_coding")

    val inFileMD5DF = inFileMD5RDD.toDF(colIdx, colData, colMd5)
    //
    ////    inFileMD5DF.show()
    //
    //    //fieldIndex:
    //    /**
    //      * 返回给定字段值的索引
    //      *
    //      * 当模板未定义时抛出：UnsupportedOperationException
    //      * 当字段名不存在时：throws IllegalArgumentException
    //      *
    //      */
    //
    //
    val crossData = inFileMD5DF
      .limit(limitCnt)
      .join(sqlFullData, colMd5) //?
      .map(x => {
      (x.getLong(x.fieldIndex(colIdx)), x.getString(x.fieldIndex(colData)), x.getString(x.fieldIndex(colMd5)))
    }).toDF(colIdx, colData, colMd5)
    //
    crossData.foreach(println)
    //
    val num = crossData.count()
    //
    //    //        crossData.show(numRows = 100)
    //
    //    //    val frame = inFileMD5DF.limit(limitCnt).select(colIdx).except(crossData.select(colIdx))
    //
    //    //    frame.foreach(println)
    //
    //
    if (num != limitCnt) {
      val diffMd5Rdd = inFileMD5DF.join(
        //except 通过对部分行抽样返回一个新的DF:
        inFileMD5DF.limit(limitCnt).select(colIdx).except(crossData.select(colIdx)),
        colIdx
      ).rdd
        .sortBy(x => x.getLong(x.fieldIndex(colIdx)))
        .map(x => x.getString(x.fieldIndex(colData)))
      //
      //      //      val schema = StructType(colNames.map(fieldName => StructField(fieldName, StringType, true)))
      //      //      val diffMd5Ds = spark.createDataFrame(diffMd5Rdd, schema)
      //      //      diffMd5Ds.show()
      //      //      diffMd5Ds.toString().foreach(println)
      //
      //      println("进入此流程!")
      //
      ////      diffMd5Ds.show()
      //
      ////      val outputDir = "spark/data/result"
      ////
      ////      diffMd5Ds.write.mode(SaveMode.Overwrite).text(outputDir.toString)
      //    }
      println("*" * 100)
      //      diffMd5Ds.foreach(println).toString
      //
      //      val source = Source.fromFile("F:\\Sjc\\code\\myStudentCode\\Spark\\data\\result.log")
      //
      //      source

      val writer = new PrintWriter(new File("F:\\Sjc\\code\\myStudentCode\\Spark\\data\\result.log"))

      //      diffMd5Rdd.toDebugString
      //      for(i <- 1 to 10)

      writer.println("1") // 这里是自动换行 不用再加\n 切记切记
      for (elem <- diffMd5Rdd.toLocalIterator) {
        writer.append(elem)
      }

      //      writer.println(diffMd5Ds.foreach)
      //      coalesce
      //      diffMd5Ds.coalesce(1, true).saveAsTextFile("F:\\Sjc\\code\\myStudentCode\\Spark\\data\\result.log")
      writer.close()



      //      println(diffMd5Ds.toString())
    }

  }

  class Md5Verifier(removeSpace: Boolean = true, delimiter: String = "") extends Serializable {
    var _delimiter: String = delimiter
    var _removeSpace: Boolean = removeSpace

    def md5Verify(content: String): String = {
      _md5Verify(trim(content))
    }

    def trim(content: String): String = {
      var str = content
      if (!_delimiter.isEmpty) {
        str = removeStr(str, _delimiter)
      }
      if (_removeSpace) {
        str = removeStr(str, " ")
      }
      str
    }

    private def _md5Verify(content: String): String = {
      val md5 = MessageDigest.getInstance("MD5")
      val dataMd5 = md5.digest((content).getBytes)
      dataMd5.map("%02x".format(_)).mkString
    }

    def replaceDouble(content: String): String = {
      var str = content
      if (content.indexOf(".") > 0) {
        str = content.replaceAll("0+?$", "").replaceAll("[.]$", "")
      }
      str
    }

    private def removeStr(content: String, removeStr: String): String = {
      return content.replace(removeStr, "")
    }
  }


}
